package org.artifact.core.plugin.rocketmq;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;
import java.util.UUID;

/**
 * Created by xy on 2018/11/16.
 */
public class ConsumerTest {
    public static void main(String[] args) {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("acg");
        consumer.setInstanceName(UUID.randomUUID().toString());
        consumer.setConsumeMessageBatchMaxSize(32);
        consumer.setNamesrvAddr("192.168.0.179:9876");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
//        consumer.registerMessageListener(new MessageListenerConcurrently() {
//            @Override
//            public ConsumeConcurrentlyStatus consumeMessage(
//                    List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
//                for(MessageExt me : list) {
//                    if("Tag1".equals(me.getTags())){
//                        System.out.println("处理 Tag1 业务");
//                        System.out.println(new String(me.getBody()) + "消费成功" + "\n");
//                    }else if("Tag2".equals(me.getTags())){
//                        System.out.println("处理 Tag2 业务");
//                        System.out.println(new String(me.getBody()) + "消费成功" + "\n");
//                    }else if("Tag3".equals(me.getTags())){
//                        System.out.println("处理 Tag3 业务");
//                        System.out.println(new String(me.getBody()) + "消费失败" + "\n");
//                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
//                    }else{
//                        //consumer.subscribe("qch_20170706", "Tag1||Tag2||Tag3");
//                        System.out.println("过滤掉的业务"+ me.getMsgId());
//                    }
//                }
//                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
//            }
//        });
        // 设置一个Listener，主要进行消息的逻辑处理
        consumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
                consumeOrderlyContext.setAutoCommit(true);
                for (MessageExt messageExt : list) {
                    System.out.println(messageExt.getUserProperty("orderId") + " " + new String(messageExt.getBody()));
                }
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
        try {
            consumer.subscribe("CenterServer", "*");
            consumer.start();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}